package timeandwindow;

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import util.SourceUtil;

import java.time.Duration;
import java.util.HashSet;

/**
 *
 * 需求：统计每10秒内的PV/UV
 *
 * 说明：在电商网站中，PV(页面浏览量)和UV(独立访客数)是非常重要的两个流量指标
 * PV: 统计所有点击量（数据加载一次记作一次1）
 * UV:对用户id去重后得到的值（累加器：hashset）
 * PV/UV:人均重复访问量，即平均每个用户会访问多少次页面，可在一定程度上代表用户的粘度
 */
public class Flink08_PVUV {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);
         SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                 .assignTimestampsAndWatermarks(
                         WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)//延迟时间设为0
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );

         ds.print("INPUT");

         //todo 窗口：非按键分区事件时间滚动窗口
         //开窗，增量聚合，打印输出
         ds
                 .windowAll(
                         TumblingEventTimeWindows.of(Time.seconds(10))
                 )
                 .aggregate(
                         new AggregateFunction<Event, Tuple2<Long, HashSet<String>>, Double>() {
                             @Override
                             public Tuple2<Long, HashSet<String>> createAccumulator() {
                                 return new Tuple2<>(0L, new HashSet<>());//初始化要赋值0，可能出现空指针异常
                             }

                              //2.进行累加操作，每条数据调用一次
                             @Override
                             public Tuple2<Long, HashSet<String>> add(Event event, Tuple2<Long, HashSet<String>> accumulator) {

                                 accumulator.f1.add(event.getUser());
                                 accumulator.f0 +=1;
                                 return Tuple2.of(accumulator.f0, accumulator.f1);

                             }

                             @Override
                             public Double getResult(Tuple2<Long, HashSet<String>> accumulator) {
                                 return (double)accumulator.f0/accumulator.f1.size();
                             }

                             @Override
                             public Tuple2<Long, HashSet<String>> merge(Tuple2<Long, HashSet<String>> longHashSetTuple2, Tuple2<Long, HashSet<String>> acc1) {
                                 return null;
                             }
                         }
                 )
                 .print("pv/uv");


         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}
